package com.lty.mq.demo;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * RabbitMQ 异步发布确认模式生产者示例
 * 功能：发送消息并监听Broker的确认（Ack/Nack），支持批量确认和失败重发
 */
public class AsyncConfirmProducer {

    // 定义交换机名称
    private final static String EXCHANGE_NAME = "async-confirm-exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接工厂并配置RabbitMQ服务器地址
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // 设置RabbitMQ服务地址（默认localhost）
        // factory.setUsername("guest"); // 默认用户名
        // factory.setPassword("guest"); // 默认密码
        // 2. 建立连接和通道
        Connection connection = factory.newConnection(); // 创建TCP连接
        Channel channel = connection.createChannel();    // 创建通信通道
        // 3. 声明交换机（如果不存在则自动创建）
        channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 使用direct类型交换机
        // 4. 初始化批量发送参数
        int batchCount = 100;       // 总消息数量
        long msgCount = 1;          // 已发送消息计数器
        SortedSet<Long> confirmSet = new TreeSet<>(); // 存储已发送但未确认的消息序列号

        // 5. 开启发布确认模式（异步确认）
        channel.confirmSelect();

        // 6. 添加确认监听器（处理Broker的Ack/Nack响应）
        channel.addConfirmListener(new ConfirmListener() {
            /**
             * 处理消息成功确认（Broker收到消息）
             * @param deliveryTag 消息序列号（唯一标识）
             * @param multiple 是否批量确认
             */
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("✅ Ack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                if (multiple) {
                    // 批量确认：移除所有小于deliveryTag的未确认消息
                    confirmSet.headSet(deliveryTag + 1).clear(); // +1是为了包含deliveryTag本身
                } else {
                    // 单条确认：仅移除当前消息
                    confirmSet.remove(deliveryTag);
                }
            }

            /**
             * 处理消息失败确认（Broker未收到消息）
             * @param deliveryTag 消息序列号
             * @param multiple 是否批量失败
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("❌ Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                if (multiple) {
                    // 批量失败：移除所有小于deliveryTag的未确认消息（这些消息需要重发）
                    SortedSet<Long> failedSet = confirmSet.headSet(deliveryTag + 1);
                    resendMessages(channel, failedSet); // 重发失败消息
                    failedSet.clear();
                } else {
                    // 单条失败：仅重发当前消息
                    confirmSet.remove(deliveryTag);
                    resendMessage(channel, deliveryTag);
                }
            }

            /**
             * 重发单条失败消息
             */
            private void resendMessage(Channel channel, long deliveryTag) throws IOException {
                String message = "async confirm test (Resent: SeqNo=" + deliveryTag + ")";
                channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                System.out.println("♻️ Resent message with SeqNo: " + deliveryTag);
            }

            /**
             * 重发批量失败消息
             */
            private void resendMessages(Channel channel, SortedSet<Long> failedSet) throws IOException {
                for (Long seqNo : failedSet) {
                    String message = "async confirm test (Batch Resent: SeqNo=" + seqNo + ")";
                    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                }
                System.out.println("♻️ Resent " + failedSet.size() + " failed messages in batch");
            }
        });

        // 7. 循环发送消息
        while (msgCount <= batchCount) {
            long nextSeqNo = channel.getNextPublishSeqNo(); // 获取下一条消息的序列号
            String message = "async confirm test (SeqNo=" + nextSeqNo + ")";
            // 发送消息（不指定routingKey，使用空字符串）
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            confirmSet.add(nextSeqNo); // 记录待确认消息
            System.out.println("📤 Sent message with SeqNo: " + nextSeqNo);
            msgCount++; // 递增计数器
        }

        // 8. 等待所有消息确认（可选：实际项目中建议设置超时机制）
        while (confirmSet.size() > 0) {
            try {
                Thread.sleep(100); // 短暂等待确认完成
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }

        // 9. 关闭资源（实际项目中建议放在finally块或使用try-with-resources）
        channel.close();
        connection.close();
        System.out.println("🔚 All messages processed. Connection closed.");
    }
}